﻿
using System;
using System.Threading;
using System.Threading.Tasks.Dataflow;

namespace WalkthroughTDF
{
    static class TestBatchBlock
    {
        #region [************************* internal classes *************************]

        private sealed class Producer
        {
            private readonly string m_name;
            private readonly int m_prodInterval;
            private readonly int m_total;
            private BufferBlock<string> m_queue;

            public Producer(string name, int prodInterval, int total)
            {
                m_name = name;
                m_prodInterval = prodInterval;
                m_total = total;

                m_queue = new BufferBlock<string>();
            }

            public BufferBlock<string> OutQueue
            {
                get { return m_queue; }
            }

            public void Start()
            {
                for (int index = 0; index < m_total; index++)
                {
                    m_queue.Post(string.Format("{0}.{1}", m_name, index));
                    Thread.Sleep(m_prodInterval);
                }//for
            }
        }// class: Producer

        #endregion

        #region [************************* test methods *************************]

        private static Producer[] CreateProducers()
        {
            const int TOTAL = 5;
            return new Producer[] {
                new Producer("fast", 100, TOTAL), 
                new Producer("middle", 300, TOTAL),
                new Producer("slow", 500, TOTAL),
            };
        }

        private static void RunProducers(Producer[] producers, BatchBlock<string> batchQueue)
        {
            ActionBlock<string[]> actBlock = new ActionBlock<string[]>(strs =>
            {
                foreach (var str in strs)
                {
                    Console.WriteLine(str);
                }
                Console.WriteLine();
            });
            batchQueue.LinkTo(actBlock);

            foreach (var producer in producers)
            {
                producer.OutQueue.LinkTo(batchQueue);
                producer.Start();
            }
        }

        /// <summary>
        /// batch is triggered as soon as enough items are collected
        /// </summary>
        /// <param name="greedy">whether the BatchBlock should run greedy or not</param>
        private static void TestBatchOnSize(bool greedy)
        {
            var producers = CreateProducers();

            BatchBlock<string> batchQueue = new BatchBlock<string>(
                producers.Length,
                new GroupingDataflowBlockOptions { Greedy = greedy });

            RunProducers(producers, batchQueue);

            Console.ReadLine();
        }

        private static void TestTimeredBatch()
        {
            var producers = CreateProducers();

            // if you want to trigger the batch queue based on the timer, other than on the size 
            // chekanote: both conditions listed below must be satisfied:
            // 1. the batch size should be larger than number of producers, so it can never collect enough items
            // 2. the BatchBlock must be non-greedy
            BatchBlock<string> batchQueue = new BatchBlock<string>(
                producers.Length + 1, new GroupingDataflowBlockOptions { Greedy = false });

            RunProducers(producers, batchQueue);

            // start the timer
            var timer = new System.Timers.Timer { Interval = 1000 };
            timer.Elapsed += (sender, evtargs) =>
                                 {
                                     Console.WriteLine(DateTime.Now.ToLongTimeString());
                                     batchQueue.TriggerBatch();
                                 };
            timer.Start();

            Console.ReadLine();
            timer.Dispose();
        }

        public static void TestMain()
        {
            // TestBatchOnSize(true);
            // TestBatchOnSize(false);
            TestTimeredBatch();
        }// TestMain
        #endregion
    }
}
